package com.shujia.flink.tf


import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
object Demo03Window {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    /**
     * 需求：求最近10秒内单词的数量（分组）
     */
    val lineDS: DataStream[String] = env.socketTextStream("master", 8888)


    val keyDS: KeyedStream[(String, Int), String] = lineDS.flatMap(_.split(","))
      .map((_, 1))
      .keyBy(_._1)

    /**
     * window:      每隔一段时间计算一次的需求需要使用窗口计算
     *
     * TumblingProcessingTimeWindows  处理时间的滚动窗口
     *
     *
     */


    val windowDS: WindowedStream[(String, Int), String, TimeWindow] = keyDS.
      window(TumblingProcessingTimeWindows.of(Time.seconds(10)))



 //窗口内聚合计算

    val countDs: DataStream[(String, Int)] = windowDS.sum(1)




      countDs.print()

    env.execute()
  }

}
